package core;

import java.util.concurrent.ConcurrentLinkedQueue;

import elements.CombinedTuple;
import elements.Combiner;
import elements.Sorter;
import elements.Spliter;
import elements.Tuple;

public class WordCounter {

	public static int maxWorkers;
	public static int aliveWorkers = 0;
	
	public static void main(String[] args) {
		maxWorkers = Integer.parseInt(args[0]);
		String fileName = args[1];
		Spliter spliter = new Spliter(fileName);
		Sorter sorter = new Sorter();
	
		ConcurrentLinkedQueue<Tuple> maps = new ConcurrentLinkedQueue<Tuple>();
		ConcurrentLinkedQueue<Tuple> reduces = new ConcurrentLinkedQueue<Tuple>();
		
		String nextBlock = spliter.getNextBlock();
		while (nextBlock != null) {
			MapWorker mapWorker = new MapWorker(fileName, nextBlock, maps);
			mapWorker.start();
			System.out.println("Alive Map Workers: " + aliveWorkers);
			waitWorker();
			nextBlock = spliter.getNextBlock();
		}
		
		Combiner combiner = new Combiner(maps);
		CombinedTuple nextCombinedTuple = combiner.getNextCombionation();
		
		while (nextCombinedTuple != null) {
			ReduceWorker reduceWorker = new ReduceWorker(nextCombinedTuple.key, 
												nextCombinedTuple.tuples, reduces);
			reduceWorker.start();
			System.out.println("Alive Reduce Workers: " + aliveWorkers);
			waitWorker();	
			nextCombinedTuple = combiner.getNextCombionation();
		}
		
		sorter.sortAndStore(reduces);
	}

	private static void waitWorker() {
		while (aliveWorkers == maxWorkers);
	}
}
